package com.wan.kafkastreaming;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

/**
 * @author wanfeng
 * @date 2021/3/13 18:44
 */
public class LogProcessor implements Processor<byte[],byte[]> {
	private ProcessorContext context;

	@Override
	public void init(ProcessorContext processorContext) {
		this.context = processorContext;
	}

	@Override
	public void process(byte[] dummy, byte[] line) {
		String input = new String(line);

		if(input.contains("PRODUCT_RATING_PREFIX:")){
			System.out.println("book rating coming!!!!" + input);
			input = input.split("PRODUCT_RATING_PREFIX:")[1].trim();
			context.forward("logProcessor".getBytes(), input.getBytes());
		}
	}

	@Override
	public void punctuate(long l) {

	}

	@Override
	public void close() {

	}
}
